package com.bw.yk08;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.sun.org.apache.bcel.internal.generic.BasicType;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;


public class MyDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector collector) throws Exception {

        JSONObject jsonObject=new JSONObject();

        String topic=sourceRecord.topic();
        String [] splicStr=topic.split("\\.");

        String database=splicStr[1];
        String table=splicStr[2];

        Struct struct= (Struct) sourceRecord.value();


        String op=struct.getString("op");

        JSONObject data=new JSONObject();
        Struct after=struct.getStruct("after");
        if(after!=null)
        {
            Schema schema=after.schema();

            for(Field f:schema.fields())
            {
                String name=f.name();
                Object v=after.get(name);
                v=checkType(name,v);
                data.put(name,v);
            }
        }


        jsonObject.put("db",database);
        jsonObject.put("table",table);
        jsonObject.put("op",op);
        jsonObject.put("data",data);

        collector.collect(jsonObject.toJSONString());

    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(java.lang.String.class);
    }


    public Object checkType(String name,Object v)
    {
        if(name!=null&&v!=null&&name.endsWith("_time")&&v instanceof Long)
        {
            java.sql.Timestamp time =new Timestamp((Long) v);
            SimpleDateFormat sf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            return sf.format(time);
        }
        return v;
    }
}
